<?php
namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Log;
use Ratchet\Client\WebSocket;
use React\EventLoop\Loop;
use React\EventLoop\TimerInterface;
use React\Socket\Connector as ReactConnector;
use Ratchet\Client\Connector as RatchetConnector;
use Clue\React\Redis\Factory as RedisFactory;
use Clue\React\Redis\Client as RedisClient;
class WebSocketListener extends Command
{
    protected $signature = 'websocket:listen';

    protected $description = 'Listen to WebSocket server and save data to cache';
    protected $okx_websocket_url = "wss://ws.okx.com:8443/ws/v5/public";

    protected array $activeSubscriptions = [];

    protected const HEARTBEAT_INTERVAL = 25; // N < 30 秒
    protected const HEARTBEAT_TIMEOUT = 5;    // 等待 pong 的超时时间
    protected $pingTimer;
    protected $pongTimeoutTimer;
    protected $conn;
    protected $reconnectAttempts = 0;

    public function handle()
    {
        $loop = Loop::get();

        $reactConnector = new ReactConnector($loop, [
            'dns'     => '8.8.8.8',
            'timeout' => 10,
        ]);

        $ratchetConnector  = new RatchetConnector($loop, $reactConnector);

        $this->connectWebSocket($loop, $ratchetConnector);

        $this->registerSignals($loop);

        $loop->run();
    }

    private function connectWebSocket($loop, $connector)
    {
        $connector($this->okx_websocket_url)->then(
            function (WebSocket $conn) use ($loop, $connector) {
                $this->conn = $conn;

                $this->reconnectAttempts = 0;

                $this->info("✅ Connected to WebSocket server");

                // 重新订阅已存储的频道
                if (!empty($this->activeSubscriptions)) {
                    $this->info("🔄 Restoring previous subscriptions...");
                    foreach ($this->activeSubscriptions as $subscription) {
                        $subscribeMessage = json_encode([
                            'op' => 'subscribe',
                            'args' => [$subscription]
                        ]);
                        $conn->send($subscribeMessage);
                        $this->info("🔄 Restored subscription: " . $subscription['channel'] . '-' . $subscription['instId']);
                    }
                }

                $this->setupRedisSubscription($loop, $conn);

                $this->setupHeartbeat($loop, $conn);

                // 接收消息
                $conn->on('message', function ($msg) use ($loop, $conn) {
                    $this->handleMessage($msg, $loop, $conn);
                });

                // 连接关闭时重连
                $conn->on('close', function ($code = null, $reason = null) use ($loop, $connector) {
                    $this->handleClose($code, $reason, $loop, $connector);
                });
            },
            function (\Throwable $e) use ($loop, $connector) {
                $this->handleConnectionError($e, $loop, $connector);
            }
        );
    }

    private function setupRedisSubscription($loop, WebSocket $conn)
    {
        $redisFactory = new RedisFactory($loop);

        // 配置 Redis 连接（根据实际修改）
        $redisUri = sprintf(
            '%s:%d?password=%s',
            config('database.redis.default.host'),
            config('database.redis.default.port'),
            config('database.redis.default.password')
        );

        $redisFactory->createClient($redisUri)->then(

            function (RedisClient $redis) use ($conn) {

                $redis->on('connect', function () {
                    $this->info('✅ Redis 连接成功');
                });

                // 订阅消息发送频道
                $redis->subscribe('websocket_send_channel')->then(
                    function () {
                        $this->info('🎯 Redis 订阅成功');
                    },
                    function (\Exception $e) {
                        $this->error("订阅失败: " . $e->getMessage());
                    }
                );

                $redis->on('subscribe', function ($channel) {
                    $this->info("📤 Redis订阅成功: " . $channel);
                });

                // 监听订阅消息
                $redis->on('message', function ($channel, $message) use ($conn) {
                    $this->info("📤 Redis触发发送: " . substr($message, 0, 50));

                    $messageData = json_decode($message, true);

                    if ($messageData && isset($messageData['op']) && isset($messageData['args'])) {
                        foreach($messageData['args'] as $arg) {
                            $key = $arg['channel'] . '-' . $arg['instId'];
                            if($messageData['op'] == 'subscribe') {
                                $this->activeSubscriptions[$key] = $arg;
                                $this->info("📝 Stored subscription: " . $key);
                            } else if($messageData['op'] == 'unsubscribe') {
                                unset($this->activeSubscriptions[$key]);
                                $this->info("🗑️ Removed subscription: " . $key);
                            }
                        }
                    }

                    $conn->send($message);
                });

                // 错误处理
                $redis->on('error', function (\Exception $e) {
                    $this->error("Redis错误: " . $e->getMessage());
                });
            },

            function (\Exception $e) {
                $this->error("Redis连接失败: " . $e->getMessage());
            }

        );
    }

    /**
     * @notes 发送心跳包
     * @param $loop
     * @param $conn
     * @return void
     * @author Evil丿
     * @date 2025/03/17 14:54
     */
    private function setupHeartbeat($loop, $conn)
    {
        $this->pingTimer = $loop->addTimer(
            self::HEARTBEAT_INTERVAL,
            function () use ($loop, $conn) {
                $this->sendPingAndWaitPong($loop, $conn, 'first');
            }
        );
    }

    /**
     * @notes 重置心跳时间
     * @param $loop
     * @param $conn
     * @return void
     * @author Evil丿
     * @date 2025/03/17 14:54
     */
    private function resetPingTimer($loop, $conn)
    {
        // 清除旧定时器
        if ($this->pingTimer instanceof TimerInterface) {
            $loop->cancelTimer($this->pingTimer);
        }

        // 重置心跳定时器
        $this->pingTimer = $loop->addTimer(
            self::HEARTBEAT_INTERVAL,
            function () use ($loop, $conn) {
                $this->sendPingAndWaitPong($loop, $conn,'reset');
            }
        );
    }

    /**
     * @notes 发送ping并等待pong响应
     * @param $loop
     * @param $conn
     * @param $type
     * @return void
     * @author Evil丿
     * @date 2025/03/17 14:54
     */
    private function sendPingAndWaitPong($loop, $conn, $type)
    {
        try {
            $conn->send('ping');

            $this->info("🫀 Sent ping ". $type);

            $this->pongTimeoutTimer = $loop->addTimer(
                self::HEARTBEAT_TIMEOUT,
                function () use ($conn) {
                    $this->error("⌛ Pong timeout, reconnecting...");
                    $conn->close();
                }
            );

        } catch (\Exception $e) {
            $this->error("Ping failed: " . $e->getMessage());
            $conn->close();
        }
    }

    /**
     * @notes 获取响应的数据，如果为pong则重置心跳时间，其他则处理结果
     * @param $msg
     * @param $loop
     * @param $conn
     * @return void
     * @author Evil丿
     * @date 2025/03/17 14:56
     */
    private function handleMessage($msg, $loop, $conn)
    {
        $message = $msg -> getPayload();
        // 重置心跳定时器
        $this->resetPingTimer($loop, $conn);

        // 处理 pong 响应
        if ($message === 'pong') {
            $this->info("🫀 Received pong");
            if ($this->pongTimeoutTimer instanceof TimerInterface) {
                $loop->cancelTimer($this->pongTimeoutTimer);
            }
            return;
        }

        // 业务逻辑处理
        $this->processMessage($conn, $message);
    }

    /**
     * @notes 处理结果
     * @param $conn
     * @param $message
     * @return void
     * @author Evil丿
     * @date 2025/03/17 14:56
     */
    private function processMessage($conn, $message)
    {
        try {
            $data = json_decode($message, true, 512, JSON_THROW_ON_ERROR);

            $this->info("📩 Received message");

            if(!key_exists('event', $data)) {
                $key = 'okx-channel' . $data['arg']['channel'].'-'.$data['arg']['instId'];
                Cache::put($key, $data, 3600);
                $this->info($message);
            } else {
                switch($data['event'])
                {
                    case 'error' :
                        $this->info("\n🛑 Gracefully stopping，Code:".$data['code'].",Msg:".$data['msg']);
                        $conn->close();
                        exit(0);
                        break;

                    case 'subscribe' :
                        $this->info("\n✅ Subscribe Success，channel:".$data['arg']['channel']);
                        break;

                    case 'unsubscribe' :
                        $this->info("\n❌ Unsubscribe Success，channel:".$data['arg']['channel']);

                        $key = 'okx-channel' . $data['arg']['channel'];

                        Cache::delete($key);
                        break;
                }
            }
        } catch (\JsonException $e) {
            Log::error('Invalid JSON message', ['error' => $e->getMessage()]);
        } catch (\ErrorException $e) {
            Log::error('Invalid JSON message', ['error' => $e->getMessage()]);
        }
    }

    /**
     * @notes 连接关闭
     * @param $code
     * @param $reason
     * @param $loop
     * @param $connector
     * @return void
     * @author Evil丿
     * @date 2025/03/17 14:57
     */
    private function handleClose($code, $reason, $loop, $connector)
    {
        Log::warning('❌ Connection closed. Code: '.$code.', Reason: '.$code);

        $this->error("❌ Connection closed. Code: $code, Reason: $reason");

        // 清理定时器
        if ($this->pingTimer) {
            $loop->cancelTimer($this->pingTimer);
        }

        if ($this->pongTimeoutTimer) {
            $loop->cancelTimer($this->pongTimeoutTimer);
        }

        // 指数退避重连
        $delay = min(30, pow(2, $this->reconnectAttempts));

        $this->reconnectAttempts++;

        $this->info("⏳ Reconnecting in {$delay}s...");

        $loop->addTimer($delay, function () use ($loop, $connector) {
            $this->connectWebSocket($loop, $connector);
        });
    }

    /**
     * @notes 链接失败
     * @param $e
     * @param $loop
     * @param $connector
     * @return void
     * @author Evil丿
     * @date 2025/03/17 14:57
     */
    private function handleConnectionError($e, $loop, $connector)
    {
        $this->error("⛔ Connection failed: " . $e->getMessage());

        $this->handleClose(-1, 'connection error', $loop, $connector);
    }

    private function registerSignals($loop)
    {
        // 仅在支持 pcntl 的环境处理信号
        if (extension_loaded('pcntl') && defined('SIGINT')) {
            // 处理 CTRL+C 和终止信号
            $loop->addSignal(SIGINT, function () use ($loop) {
                $this->info("\n🛑 Gracefully stopping...");
                $loop->stop();
                exit(0);
            });

            $loop->addSignal(SIGTERM, function () use ($loop) {
                $this->info("\n🛑 Terminating...");
                $loop->stop();
                exit(0);
            });
        } else {
            // Windows 或未安装 pcntl 的替代方案
            $this->warn('⚠️  Signal handling (SIGINT/SIGTERM) is not supported in this environment.');

            $this->warn('⚠️  Press CTRL+C to stop manually.');
        }
    }
}
